package jwebtest.rpc.zk;

import java.util.ArrayList;
import java.util.List;
import java.util.Scanner;
import java.util.concurrent.CountDownLatch;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

import com.jweb.util.FormatUtil;

/**
 * ZooKeeper Java Api 使用样例<br>
 * ZK Api Version: 3.4.3
 */
public class ZkApi implements Watcher {

	private ZooKeeper zk = null;
	private CountDownLatch latch = new CountDownLatch(1);

	public ZooKeeper getZK() {
		return zk;
	}

	/**
	 * 创建ZK连接
	 * 
	 * @param connectString
	 *            ZK服务器地址列表
	 * @param sessionTimeout
	 *            Session超时时间
	 */
	public void createConnection(String connectString, int sessionTimeout) {
		this.close();
		try {
			zk = new ZooKeeper(connectString, sessionTimeout, this);
			latch.await();
		} catch (Exception e) {
			System.out.println("连接创建失败,connectString=" + connectString);
			e.printStackTrace();
		}
	}

	/**
	 * 关闭ZK连接
	 */
	public void close() {
		if (!FormatUtil.isEmpty(this.zk)) {
			try {
				this.zk.close();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

	/**
	 * 获取所有子节点
	 * 
	 * @param path
	 * @return
	 */
	public List<String> getChildren(String path) {
		try {
			return zk.getChildren(path, false);
		} catch (Exception e) {
			e.printStackTrace();
		}
		return new ArrayList<String>();
	}

	/**
	 * 创建节点
	 * 
	 * @param path
	 *            节点path
	 * @param data
	 *            初始数据内容
	 * @return
	 */
	public boolean createPath(String path, String data, CreateMode mode) {
		try {
			System.out.println("节点创建成功, Path: " + this.zk.create(path, data.getBytes(), Ids.OPEN_ACL_UNSAFE, mode) + ", content: " + data);
			return true;
		} catch (Exception e) {
			System.out.println("节点创建失败,path=" + path);
			e.printStackTrace();
		}
		return false;
	}

	public boolean exists(String path) {
		try {
			Stat stat = this.zk.exists(path, false);
			System.out.println("判断节点是否存在: stat=" + stat);
			return stat == null ? false : true;
		} catch (Exception e) {
			System.out.println("判断节点是否存在失败，path=" + path);
			e.printStackTrace();
		}
		return false;
	}

	/**
	 * 读取指定节点数据内容
	 * 
	 * @param path
	 *            节点path
	 * @return
	 */
	public String readData(String path) {
		try {
			System.out.println("获取数据成功，path：" + path);
			return new String(this.zk.getData(path, false, null));
		} catch (Exception e) {
			System.out.println("读取数据失败，path: " + path);
			e.printStackTrace();
			return "";
		}
	}

	/**
	 * 更新指定节点数据内容
	 * 
	 * @param path
	 *            节点path
	 * @param data
	 *            数据内容
	 * @return
	 */
	public boolean writeData(String path, String data) {
		try {
			System.out.println("更新数据成功，path：" + path + ", stat: " + this.zk.setData(path, data.getBytes(), -1));
			return true;
		} catch (Exception e) {
			System.out.println("更新数据失败，path: " + path);
			e.printStackTrace();
		}
		return false;
	}

	/**
	 * 删除指定节点
	 * 
	 * @param path
	 *            节点path
	 */
	public void deleteNode(String path) {
		try {
			this.zk.delete(path, -1);
			System.out.println("删除节点成功，path：" + path);
		} catch (Exception e) {
			System.out.println("删除节点失败，path: " + path);
			e.printStackTrace();
		}
	}

	/**
	 * 收到来自Server的Watcher通知后的处理。
	 */
	@Override
	public void process(WatchedEvent event) {
		System.out.println("收到事件通知：" + event.getState() + "\n");
		if (KeeperState.SyncConnected == event.getState()) {
			latch.countDown();
		}
	}

	public static void main(String[] args) throws KeeperException, InterruptedException {
		int SESSION_TIMEOUT = 10000;
		String CONNECTION_STRING = "192.168.1.23:2181";
		// =============================================
		ZkApi sample = new ZkApi();
		sample.createConnection(CONNECTION_STRING, SESSION_TIMEOUT);
		boolean bo = true;
		while (bo) {
			forEach("/", sample.getZK(), -1);
			System.out.print("选择操作:\n1.刷新\t2.删除\t9.退出\n");
			int ind = FormatUtil.toInt(new Scanner(System.in).next());
			switch (ind) {
			case 2:
				System.out.println("请输入节点名称：\n");
				String temp = new Scanner(System.in).next();
				sample.deleteNode(temp);
				break;
			case 9:
				bo = false;
			default:
				break;
			}
		}
		sample.close();
	}

	public static void forEach(String path, ZooKeeper zk, int dep) throws KeeperException, InterruptedException {
		List<String> list = zk.getChildren(path, false);
		++dep;
		if (list.size() > 0)
			for (String str : list) {
				String temp = FormatUtil.formatPath(path + "/" + str);
				byte[] data = zk.getData(temp, false, null);
				System.err.println("[" + dep + "]" + temp + "[value=" + new String(data == null ? new byte[] {} : data) + "]");
				forEach(temp, zk, dep);
			}
	}
}